package com.doitedu.mr.day04.movie;

import com.alibaba.fastjson.JSON;
import com.doitedu.mr.day04.bean.MovieWriable;
import com.doitedu.mr.day04.json.Movie;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @Date 2021/12/3
 * @Created by HANGGE
 * @Description TODO
 */
public class AvgRate02 {
    //Mapper  K : movieID V: rate
    // 将自定义的类放在泛型中  自定义的类要实现HDP的序列化
    static  class AvgRateMapper extends Mapper<LongWritable , Text , Text, MovieWriable>{
        Text k = new Text() ;
        // 每行数据执行一次
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            try {
                String line = value.toString();
                //将行数据转换成Movie
                MovieWriable movie = JSON.parseObject(line, MovieWriable.class);
                // 获取电影的ID  分数
                String id = movie.getMovie();
                // 将数据封装起来
                k.set(id);
                context.write(k,movie);
               // context.write(new Text(id) , new DoubleWritable(rate));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    //Reducer
    static  class AvgRateReducer extends Reducer<Text , MovieWriable , Text , DoubleWritable>{
        DoubleWritable v = new DoubleWritable() ;
        @Override
        protected void reduce(Text key, Iterable<MovieWriable> values, Context context) throws IOException, InterruptedException {
            double sum = 0d  ;
            int cnt = 0 ;

            for (MovieWriable value : values) {
                double rate = value.getRate();
                 sum+=rate ;
                 cnt++ ;
            }

            double avg = (sum/cnt) ;
            v.set(avg);
            context.write(key , v);

        }
    }

    //main
    public static void main(String[] args)throws  Exception {
        // Job  统计文件中单词出现的次数
        Configuration conf = new Configuration();
        // 1  创建 JOB
        Job job = Job.getInstance(conf, "wordcount");
        // 1) Mapper 类
        job.setMapperClass(AvgRateMapper.class);
        // 2) Reducer类
        job.setReducerClass(AvgRateReducer.class);
        // 3)  设置map输出的数据类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(MovieWriable.class);
        // 4)  设置reduce输出的数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
        // 5) 设置输入的数据路径
        FileInputFormat.setInputPaths(job , new Path("E:\\mrdata\\movie\\input"));
        // 6) 设置输出的结果保存路径
        FileOutputFormat.setOutputPath(job, new Path("E:\\mrdata\\movie\\avg_rate2"));
        // 2 提交工作
        // 等待执行完成 , 打印执行过
        job.waitForCompletion(true) ;
    }

}
